package com.msg.thread;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.msg.model.MsgPush;
import com.msg.utils.MsgSrvUtils;
import com.pjc.config.BaseEnums;

/**
 * 群推消息队列
 * 
 * @author 王畅
 * @since 2015年6月21日
 */
public class GroupPushThread implements Runnable {
	Logger log = LoggerFactory.getLogger(GroupPushThread.class);
	public static final int PUSH_LIST_SIZE = 2048;
	private LinkedBlockingQueue<MsgPush> list = new LinkedBlockingQueue<MsgPush>(2048);
	private static GroupPushThread instance = null;
	private static int PAGE_SIZE = 20;

	public static GroupPushThread getInstance() {
		synchronized (GroupPushThread.class) {
			if (instance == null) {
				instance = new GroupPushThread();
				new Thread(instance).start();
			}
		}
		return instance;
	}

	public void addPushMsg(MsgPush pushMsg) {
		try {
			if (list.size() < PUSH_LIST_SIZE)
				list.add(pushMsg);
		} catch (Exception e) {
			this.log.error("添加队列失败", e);
		}
	}

	public void run() {
		try {
			while (true) {
				if (list.size() > 0) {
					List<MsgPush> pushList = new ArrayList<MsgPush>();
					list.drainTo(pushList, PAGE_SIZE);
					Set<String> set = new HashSet<String>();
					for (MsgPush msg : pushList) {
						set.add(msg.getAppId());
					}
					List<String> appList = new ArrayList<String>();
					appList.addAll(set);
					for (String appID : appList) {
						JSONArray jlist = new JSONArray();
						for (MsgPush msg : pushList) {
							if (appID.equalsIgnoreCase(msg.getAppId())) {
								jlist.add(msg.getData());
							}
						}
						JSONObject returnJson = new JSONObject();
						returnJson.put("msgId", Integer.valueOf(203));
						returnJson.put("type", BaseEnums.NoticeTypeEnum.Notice_FlowData.getId());
						returnJson.put("appId", appID);
						returnJson.put("flowData", jlist);
						this.log.debug("{}滚动条消息：{}", appID, returnJson);
						MsgSrvUtils.broadcastMsg(returnJson, appID);
					}
				}
				Thread.currentThread();
				Thread.sleep(60000L);
			}
		} catch (InterruptedException e) {
			this.log.error("群推失败", e);
		} catch (Exception e) {
			this.log.error("查询群发消息失败", e);
		}
	}
}